#pragma once
#include <ev.h>
#include <amqpcpp.h>
#include <amqpcpp/libev.h>
#include <openssl/ssl.h>
#include <openssl/opensslv.h>
#include <iostream>
#include <functional>
#include "logger.hpp"

class MyMQClient
{
    using MSCallback = std::function<void(const char*, size_t)>;
public:
    MyMQClient(const std::string &user, const std::string &passwd, const std::string &host)
    {
        // 1. 实例化底层网络通信框架的I/O事件监控句柄
        _loop = EV_DEFAULT;
        // 2. 实例化libEvHandler句柄 --- 将AMQP框架与事件监控关联起来
        _handler = std::make_unique<AMQP::LibEvHandler>(_loop);
        // 2.5. 实例化连接对象
        //"amqp://root:qdqdqd@127.0.0.1:5672/"
        const std::string url = "amqp://" + user + ":" + passwd + "@" + host + "/";
        AMQP::Address address(url);
        _connection = std::make_unique<AMQP::TcpConnection>(_handler.get(), address);
        // 3. 实例化信道对象
        _channel = std::make_unique<AMQP::TcpChannel>(_connection.get());
        _loop_thread = std::thread([this]()
                                   { ev_run(_loop, 0); });

    }
    ~MyMQClient()
    {
        ev_async_init(_w, watcher_callback);
        ev_async_start(_loop, _w);
        ev_async_send(_loop, _w);
        _loop_thread.join();
        _loop = nullptr;
    }
    bool declare_bind(const std::string &exchange, const std::string &queue, 
        const std::string &routingkey="routing_key",
        AMQP::ExchangeType echange_type = AMQP::ExchangeType::direct)
    {
        // 4. 声明交换机
        _channel->declareExchange(exchange, echange_type)
            .onError([](const char *message)
                     {
                        LOG_ERROR("声明交换机失败：{}", message);
                        return false; })
            .onSuccess([exchange]()
                       { LOG_ERROR("{} 交换机创建成功！", exchange); });
        // 5. 声明队列
        _channel->declareQueue(queue)
            .onError([](const char *message)
                     {
                        LOG_ERROR("声明队列失败：{}", message);
                        return false; })
            .onSuccess([queue]()
                       { LOG_ERROR("{} 队列创建成功！", queue); });
        // 6. 针对交换机和队列进行绑定
        _channel->bindQueue(exchange, queue, routingkey)
            .onError([exchange,queue](const char *message)
                     {
                        LOG_ERROR("{} - {} - 绑定失败: {}",exchange,queue,message);
                        return false; })
            .onSuccess([exchange,queue]()
                       { LOG_ERROR("{} - {} - 绑定成功",exchange,queue); });
                       return true;
    }
    bool publish(const std::string &msg, const std::string &exchange, const std::string &routingkey="routing_key")
    {
        bool ret = _channel->publish(exchange, routingkey, msg);
        if (ret == false)
        {
            LOG_ERROR("publish 失败");
            return false;
        }
        return true;
    }
    void callback(const MSCallback& cb, const AMQP::Message &message, uint64_t deliveryTag, bool redelivered)
    {
        cb(message.body(), message.bodySize());
        _channel->ack(deliveryTag); // 对消息进行确认
    }
    bool consume(const std::string &queue, const std::string &tag, const MSCallback& cb)
    {
        auto Callback = std::bind(&MyMQClient::callback, this, cb, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3);
        _channel->consume(queue, tag) // 返回值 DeferredConsumer
            .onReceived(Callback)
            .onError([queue](const char *message)
                     {
                        LOG_ERROR("订阅 {} 队列消息失败: {}",queue,message);
                        return false; 
                    }); // 返回值是 AMQP::Deferred
    return true;
    }

private:
    static void watcher_callback(struct ev_loop *loop, ev_async *watcher, int32_t revents)
    {
        ev_break(loop, EVBREAK_ALL);
    }

private:
    struct ev_async* _w;
    struct ev_loop *_loop;
    std::unique_ptr<AMQP::LibEvHandler> _handler;
    std::unique_ptr<AMQP::TcpConnection> _connection;
    std::unique_ptr<AMQP::TcpChannel> _channel;
    std::thread _loop_thread;
};